package com.baic.ibase.config;


import com.baic.ibase.service.mqtt.ReceiverMessageHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;


/**
 * mqtt消息入站适配器
 */
@Configuration
public class MqttInboundConfiguration {
    @Autowired
    private MqttPropertiesConfig mqttPropertiesConfig;

    @Autowired
    private MqttPahoClientFactory mqttClientFactory;

    @Autowired
    private ReceiverMessageHandler receiverMessaeHandler;

    @Bean
    public MessageChannel messageInboundChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer messageProducer() {
        MqttPahoMessageDrivenChannelAdapter messageProducer = new MqttPahoMessageDrivenChannelAdapter(
                mqttPropertiesConfig.getUrl(),
                mqttPropertiesConfig.getSubClientId(),
                mqttClientFactory,
                mqttPropertiesConfig.getSubscriptionTopic().split(","));
        messageProducer.setQos(mqttPropertiesConfig.getSubscriptionQos());
        messageProducer.setConverter(new DefaultPahoMessageConverter());
        messageProducer.setOutputChannel(messageInboundChannel());
        return messageProducer;
    }


    @Bean
    @ServiceActivator(inputChannel = "messageInboundChannel")
    public MessageHandler mqttInboundHandler() {
        return receiverMessaeHandler;
    }

}
